package streaming.api.source;

import streaming.api.beans.SensorReading;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.source.ParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.RichParallelSourceFunction;
import org.apache.flink.streaming.api.functions.source.SourceFunction;

import java.util.HashMap;
import java.util.Map;
import java.util.Random;

/**
 * 从自定义源读取数据
 * env.addSource
 * PS:实现并行度只能为1的自定义DataSource implements继承 SourceFunction 接口
 */
public class Source5_1 {

    private static Long sleepTime = 10000L;//睡眠10秒

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);
        DataStream dataStream = env.addSource(new ST5Source1());
        dataStream.print();
        env.execute();
    }

    //
    public static class ST5Source1 implements SourceFunction<SensorReading> {

        private boolean running = true;//定义一个标识位，用来控制数据的产生

        @Override
        public void run(SourceContext<SensorReading> out) throws Exception {
            Random random = new Random();
            Map<String, Double> map = new HashMap<String, Double>();
            for (int i = 0; i < 10; i++) {
                map.put("sensor_" + (i + 1), 60 + random.nextGaussian() * 20);
            }
            while (running) {
                for (String id : map.keySet()) {
                    Double newTemp = map.get(id) + random.nextGaussian();
                    map.put(id, newTemp);
                    out.collect(new SensorReading(id, System.currentTimeMillis(), newTemp));
                }
                Thread.sleep(sleepTime);
            }
        }

        @Override
        public void cancel() {
            running = false;
        }
    }

}
